package com.lmm.rpc.core.transport;

import com.lmm.rpc.core.client.async.RpcCallback;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class RpcFuture<T> implements Future<T> {

    private long timeoutMillis;
    private final long beginTimestamp = System.currentTimeMillis();

    /**
     *  响应结果
     */
    private T response;

    /**
     *  异步回调接口
     */
    private RpcCallback rpcCallback;

    private final AtomicBoolean executeCallbackOnlyOnce = new AtomicBoolean(false);
    //发送成功标记
    private volatile boolean sendRequestOK = true;
    /**
     * 因为请求和响应是一一对应的，所以这里是1
     * 使用 CountDownLatch 等待线程
     */
    private CountDownLatch countDownLatch = new CountDownLatch(1);


    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    /**
     *  响应数据不为空 表示完成
     * @return
     */
    @Override
    public boolean isDone() {
        return this.response != null;
    }

    /**
     *  等待获取数据，直到有结果 也就是 countDownLatch 的值减到 0
     *
     *  使用countDownLatch.await()的原因，要先存才能取，避免拿的时候，服务数据还没返回
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @Override
    public T get() throws InterruptedException {
        // 进入阻塞等待 countDownLatch减少值为0 返回下面结果
        countDownLatch.await();
        return response;
    }

    /**
     *  超时等待 获取数据
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException {
        if (countDownLatch.await(timeout,unit)){
            return response;
        }
        return null;
    }

/*
* 设置返回的数据到RpcFuture对象*/
    public void setResponse(T response){
        this.response = response;
        countDownLatch.countDown();
    }
    /**
     * 执行回调逻辑
     */
    public void executeInvokeCallback(Throwable e) {
        if (Objects.nonNull(rpcCallback)){
            if (this.executeCallbackOnlyOnce.compareAndSet(false, true)) {
                if (sendRequestOK)
                    rpcCallback.onSuccess(this);
                else
                    rpcCallback.onException(e);
            }
        }
    }
}
